
#include <sstream>
#include "PostgreStorage.h"
#include "Define.h"
#include <pqPool.h>
#include <viva.h>
#include "Debug.h"
#include "Utils.h"
using namespace std;
using namespace heron;

bool PostgreStorage :: Initialize(const char *conf) 
{
 DEBUG2("PqPool::Instance ", conf );
	PqPool *pA =   PqPool::Instance(conf); 	
	if(pA == NULL)
        {
                ERRO <<" PqPool Instance error" << endl;
		return false;
        } 
	else
	{
/*
DEBUG1("start to check Postgre");
		int fc = pA->GetFarmsCounts();
		for(int i=0; i<fc; i++)
		{
			PGconn* rcon = pA->GetReader(i);
			if(rcon == NULL)
                        {
                                cerr<<"Get reader by farmid="<< i<<", but return is NULL" << endl;
				return false;
                        } 
		}
*/
		GKeyFile *kv = g_key_file_new();
		GError *err = NULL;
		if(TRUE != g_key_file_load_from_file(kv,conf, G_KEY_FILE_NONE , &err))
		{
			if (err != NULL)
			{
				cerr << "Unable to read file:" << err->message << endl;
				g_error_free (err);
			}
			g_key_file_free(kv);
			return false;
		}

		gchar *pval =  g_key_file_get_value(kv,"global", "writedb",  NULL);
		if(pval == NULL)
		{
			cerr<<"don't write to db!"<< endl;
			m_bWrite2DB = false;
		}
		else
		{
			atoi((char *)pval) > 0 ? m_bWrite2DB = true: m_bWrite2DB = false;  
			g_free(pval);
		}
		pval =  g_key_file_get_value(kv,"global", "txlog",  NULL);
		if(pval == NULL)
		{
			cerr<<"no txlog in global"<< endl;
                        m_pTx = NULL;
		}
		else
		{ 
			string txconf = pval;
			g_free(pval);
			g_key_file_free(kv);
			m_pTx = new  Viva();
			if(!m_pTx->Initilize(txconf.c_str()))
			{
				delete m_pTx;
                                m_pTx = NULL;
				return false;
			}
		}   
		return true;
	}
}

int PostgreStorage :: Write(int ihash, const char *uid, const char *dm, const char *key, const char *val, const char *opt) 
{
        long lt = time(NULL);
	char sbuf[128];
	int ilen = sprintf(sbuf, "%ld", lt);
	sbuf[ilen] = '\0';
	int iret = 0;
	if(m_pTx != NULL)
	{
#ifdef  _THREAD_SAFE
		pthread_mutex_lock(&_lock);
#endif
		iret = write_viva(uid, dm, key, val,(const char *)sbuf, opt);
#ifdef  _THREAD_SAFE
		pthread_mutex_unlock(&_lock);
#endif

	}
	if(m_bWrite2DB)
		return write_db(ihash, uid, dm, key, val, (const char *)sbuf, opt);   	
	return iret;
}

int PostgreStorage :: write_db(int ihash, const char *uid, const char *dm, const char *key, const char *val,const char *stime, const char *opt) 
{
	PqPool *pAn = PqPool::Instance();
	if(pAn == NULL )
		return HERON_ERR_SYSFATAL;
        ihash = ihash%(1+pAn->GetMaxKey());
	int fId = pAn->GetIDbyKey(ihash);
	PGconn  *conn = pAn->GetWriter(fId, 0);
	if(conn == NULL)
	{
		pAn->CloseWriter(fId, conn);
		cerr<<"can't get mysql reader connection!"<<endl;	
		return HERON_ERR_SYSFATAL;
	}

       	char su[ 2*HERON_KEY_MAXSIZE +1 ];
	int elen = PQescapeString(su, uid, strlen(uid));
	su[elen] = '\0';
	ostringstream stb;
	stb << HERON_DB_T_HEADER << ihash;
	bool bret = false;
	if(strcasecmp(opt, HERON_CMD_ATOM) == 0)
	{
		bret = Db_Write_Atom(conn, stb.str().c_str(),(char *)su, dm, key, val, stime);
	}
	else if(strcasecmp(opt, HERON_CMD_CREATE) == 0)
	{
/*		char eval[ strlen(val)*2+ 1 ];
		int elen = mysql_escape_string(eval, val, strlen(val));
		eval[elen] = '\0'; */
		bret = Db_Write_Create(conn, stb.str().c_str(),(char *)su, dm, HERON_UUID_KEY, val, stime);
	}
	else if(strcasecmp(opt, HERON_CMD_REPLACE) == 0)
	{
		char eval[ strlen(val)*2+ 1 ];
		int elen = PQescapeString(eval, val, strlen(val));
		eval[elen] = '\0';
		bret = Db_Write_Set(conn, stb.str().c_str(),(char *)su, dm, key, eval, stime);
	}
	else 
	{
		pAn->CloseWriter(fId, conn);
		return HERON_ERR_PARAMERR;
	}
	pAn->CloseWriter(fId, conn);
       if(bret)
            return HERON_SUCCESS;
       return HERON_ERR_SYSFATAL;
}
int PostgreStorage :: write_viva(const char *uid, const char *dm, const char *key, const char *val, const char *stime, const char *opt) 
{
        if(m_pTx == NULL)
            return HERON_ERR_SYSFATAL;
	m_pTx->BeginTX();
	m_pTx->AddData(HERON_DOMAIN_KEY, dm);
	m_pTx->AddData(HERON_USRID_KEY, uid);
	m_pTx->AddData(HERON_CMD_KEY, opt); // HERON_CMD_REPLACE);
	m_pTx->AddData(key, val);
	m_pTx->AddData(HERON_MOD_TIME, stime);
	if(m_pTx->CommitToLog())
		return HERON_SUCCESS;
	return HERON_ERR_PARAMERR;
}

int PostgreStorage :: Read(int ihash, const char *uid, const char *dm, const char *key, string &val)
{
	PqPool *pAn = PqPool::Instance();
	if(pAn == NULL )
		return HERON_ERR_SYSFATAL;
        ihash = ihash%(1+pAn->GetMaxKey());
	int fId = pAn->GetIDbyKey(ihash);
	PGconn* conn = pAn->GetReader(fId);
	if(conn == NULL)
	{
		pAn->CloseReader(fId, conn);
		cerr << __func__<<":"<<__LINE__<<":"<<" getReader("<<fId<<") == NULL"<<endl;
		return HERON_ERR_SYSFATAL;
	}
	bool qRet = read_user_value(conn, ihash, uid, dm, key, val);
	pAn->CloseReader(fId, conn);
	if(qRet)
		return HERON_SUCCESS;
	else
		return HERON_ERR_KEY_NOEXIST;
}


int PostgreStorage :: Dump(int ihash, const char *uid, const char *dm, GHashTable *kv)
{
	PqPool *pAn = PqPool::Instance();
	if(pAn == NULL )
		return HERON_ERR_SYSFATAL;
        ihash = ihash%(1+pAn->GetMaxKey());
	int fId = pAn->GetIDbyKey(ihash);
	PGconn  *conn = pAn->GetReader(fId);
	if(conn == NULL)
	{
		pAn->CloseReader(fId, conn);
		cerr<<"can't get mysql reader connection!"<<endl;	
		return HERON_ERR_SYSFATAL;
	}
        g_hash_table_remove_all(kv);
        int iRet = 0;
	iRet = dump_user_value(conn, ihash, uid, dm, kv);
	pAn->CloseReader(fId, conn);
	if(iRet > 0 )
		return g_hash_table_size(kv);
	return HERON_ERR_USR_NOEXIST;
}

int PostgreStorage :: Read(int ihash, const char *uid, const char *dm, GList *rkv,GHashTable *kv)
{
        if(rkv == NULL)
            return HERON_SUCCESS;

	PqPool *pAn = PqPool::Instance();
	if(pAn == NULL )
		return HERON_ERR_SYSFATAL;
        ihash = ihash%(1+pAn->GetMaxKey());
	int fId = pAn->GetIDbyKey(ihash);
	PGconn  *conn = pAn->GetReader(fId);
	if(conn == NULL)
	{
		pAn->CloseReader(fId, conn);
		cerr<<"can't get mysql reader connection!"<<endl;	
		return HERON_ERR_SYSFATAL;
	}
	int iRet = read_user_value(conn, ihash,  uid, dm, rkv, kv);
	pAn->CloseReader(fId, conn);
	if(iRet >= 0 )
		return iRet;
	else
		return HERON_ERR_SYSFATAL; 
}

int PostgreStorage :: Create(int ihash, const char *uid, const char *dm, GHashTable *kv)
{
        long lt = time(NULL);
        char sbuf[128];
        int ilen = sprintf(sbuf, "%ld", lt);
        sbuf[ilen] = '\0';
        int iret = 0;
        if(m_pTx != NULL)
        {
#ifdef  _THREAD_SAFE
                pthread_mutex_lock(&_lock);
#endif
                iret = write_viva(uid, dm, kv, (const char *)sbuf, (const char *)HERON_CMD_CREATE);
#ifdef  _THREAD_SAFE
                pthread_mutex_unlock(&_lock);
#endif

        }
        if(m_bWrite2DB)
                return write_db(ihash, uid, dm, kv, (const char *)sbuf, (const char *)HERON_CMD_CREATE);
        return iret;
}

int PostgreStorage :: create_viva(const char *uid, const char *dm, const char *val ,const char *stime)
{
        DEBUG1( uid );
	m_pTx->BeginTX();
	m_pTx->AddData(HERON_CMD_KEY, HERON_CMD_CREATE);
	m_pTx->AddData(HERON_USRID_KEY, uid);
	m_pTx->AddData(HERON_DOMAIN_KEY, dm);
	m_pTx->AddData(HERON_UUID_KEY, val);
	m_pTx->AddData(HERON_MOD_TIME, stime);
	if(m_pTx->CommitToLog())
		return HERON_SUCCESS;
        DEBUG1("write txlog error!"); 
//	if(CnuppSQL::IsValidHash(yuser) && pAn->excute2TX(yuser))
	return HERON_ERR_SYSFATAL;
}

int PostgreStorage :: create_db(int ihash, const char *uid, const char *dm, const char *val, const char *stime)
{
        DEBUG1(uid);
	PqPool *pAn = PqPool::Instance();
	if(pAn == NULL )
		return HERON_ERR_SYSFATAL;
        ihash = ihash%(1+pAn->GetMaxKey());
	int fId = pAn->GetIDbyKey(ihash);
	PGconn  *conn = pAn->GetWriter(fId, 0);
	if(conn == NULL)
	{
		pAn->CloseWriter(fId, conn);
		cerr<<"can't get mysql reader connection!"<<endl;	
		return HERON_ERR_SYSFATAL;
	}

       	char su[ 2*HERON_KEY_MAXSIZE +1 ];
	int elen = PQescapeString(su, uid, strlen(uid));
	su[elen] = '\0';
   
	ostringstream stb;
	stb << HERON_DB_T_HEADER << ihash;
        
        bool bret = Db_Write_Create(conn, stb.str().c_str(),(char *)su, dm, HERON_UUID_KEY, val, stime);
	pAn->CloseWriter(fId, conn);
	if(bret )
		return HERON_SUCCESS;
	else
		return HERON_ERR_SYSFATAL; 
}

int PostgreStorage :: Write(int ihash, const char *uid, const char *dm, GHashTable *kv, const char *opt)
{
	if(g_hash_table_size(kv) <= 0 ) // && strcmp(opt, HERON_CMD_CREATE) !=0 )
		return HERON_ERR_PARAMERR;

	long lt = time(NULL);
	char sbuf[128];
	int ilen = sprintf(sbuf, "%ld", lt);
	sbuf[ilen] = '\0';
	int iret = 0;
	if(m_pTx != NULL)
	{
#ifdef  _THREAD_SAFE
		pthread_mutex_lock(&_lock);
#endif
		iret = write_viva(uid, dm, kv, (const char *)sbuf, opt);
#ifdef  _THREAD_SAFE
		pthread_mutex_unlock(&_lock);
#endif

	}
	if(m_bWrite2DB)
		return write_db(ihash, uid, dm, kv, (const char *)sbuf, opt);    
	return iret;
}

int PostgreStorage :: write_db(int ihash, const char *uid, const char *dm, GHashTable *kv,const char *stime, const char *opt)
{// opt == HERON_CMD_REPLACE
 // HERON_CMD_RM     
 // HERON_CMD_CREATE        
 // HERON_CMD_REPLACE     
 // HERON_CMD_ATOM     
	PqPool *pAn = PqPool::Instance();
	if(pAn == NULL )
		return HERON_ERR_SYSFATAL;
	ihash = ihash%(1+pAn->GetMaxKey());
	int fId = pAn->GetIDbyKey(ihash);
	PGconn  *conn = pAn->GetWriter(fId, 0);
	if(conn == NULL)
	{
		pAn->CloseWriter(fId, conn);
		cerr<<"can't get mysql reader connection!"<<endl;	
		return HERON_ERR_SYSFATAL;
	}

	char su[ 2*HERON_KEY_MAXSIZE +1 ];
	int elen = PQescapeString(su, uid, strlen(uid));
	su[elen] = '\0';
	ostringstream stb;
	stb << HERON_DB_T_HEADER << ihash;
	bool bret = false;

	if(strcasecmp(opt, HERON_CMD_ATOM) == 0)
	{
		GHashTableIter iter;
		gpointer key, value;
		g_hash_table_iter_init (&iter, kv);
		while (g_hash_table_iter_next (&iter, &key, &value))
			bret = Db_Write_Atom(conn, stb.str().c_str(),(char *)su, dm,(char *)key, (char *)value, stime);
	}
	else if(strcasecmp(opt, HERON_CMD_CREATE) == 0)
	{
		char *uv =(char *) g_hash_table_lookup(kv, HERON_UUID_KEY);
		if(uv == NULL)
			uv =(char *)"";		
		bret = Db_Write_Create(conn, stb.str().c_str(),(char *)su, dm, HERON_UUID_KEY, uv , stime);
		if( bret)
		{
			GHashTableIter iter;
			gpointer key, value;
			g_hash_table_iter_init (&iter, kv);
			while (g_hash_table_iter_next (&iter, &key, &value))
			{
                                if(strcmp((char *)key, HERON_UUID_KEY) == 0)
                                       continue; 
				char eval[ strlen((char *)value)*2+ 1 ];
				int elen = PQescapeString(eval,(char *)value, strlen((char *)value));
				eval[elen] = '\0';
				bret = Db_Write_Set(conn, stb.str().c_str(),(char *)su, dm,(char *)key, eval, stime);
			}
		}
	}
	else if(strcasecmp(opt, HERON_CMD_REPLACE) == 0)
	{
		GHashTableIter iter;
		gpointer key, value;
		g_hash_table_iter_init (&iter, kv);
		while (g_hash_table_iter_next (&iter, &key, &value))
		{
			char eval[ strlen((char *) value)*2+ 1 ];
			int elen = PQescapeString(eval,(char *)value, strlen((char *)value));
			eval[elen] = '\0';
			bret = Db_Write_Set(conn, stb.str().c_str(),(char *)su, dm,(char *)key, eval, stime);
		}
	}
	else 
	{
		pAn->CloseWriter(fId, conn);
		return HERON_ERR_PARAMERR;
	}

	pAn->CloseWriter(fId, conn);
	if(bret)
		return HERON_SUCCESS;
	return HERON_ERR_SYSFATAL;
}

int PostgreStorage :: write_viva(const char *uid, const char *dm, GHashTable *kv,const char *stime, const char *opt)
{
	m_pTx->BeginTX();
	m_pTx->AddData(HERON_CMD_KEY, opt);
	GHashTableIter iter;
	gpointer key, value;
	g_hash_table_iter_init (&iter, kv);
	while (g_hash_table_iter_next (&iter, &key, &value)) 
		m_pTx->AddData((char *)g_strdup((char *)key), (char *)g_strdup((char *)value));

	m_pTx->AddData(HERON_DOMAIN_KEY, dm);
	m_pTx->AddData(HERON_USRID_KEY, uid);
	m_pTx->AddData(HERON_MOD_TIME, stime);

	if(m_pTx->CommitToLog())
		return HERON_SUCCESS;

	return HERON_ERR_SYSFATAL;
}

int PostgreStorage :: UUID( int ihash, const char *uid)
{
 	PqPool *pAn = PqPool::Instance();
	if(pAn == NULL )
		return HERON_ERR_SYSFATAL;
        int ifarms = pAn->GetFarmsCounts();
        ihash = ihash%ifarms;

	PGconn  *conn = pAn->GetWriter(ihash, 0);
	if(conn == NULL)
	{
                pAn->CloseWriter(ihash, conn); 
		cerr<<"can't get connection!"<<endl;	
		return HERON_ERR_SYSFATAL;
	}
        ostringstream sql;
        sql<<"UPDATE guid set uuid ="<< ifarms<<" + uuid WHERE id=1; SELECT uuid FROM guid WHERE id=1";
        DEBUG1(sql.str().c_str());
        PGresult *result = PQexec(conn, sql.str().c_str());
        if(PQresultStatus(result) != PGRES_TUPLES_OK)
        {
		ERRO << PQerrorMessage(conn) << endl;
                PQclear(result); 
                pAn->CloseWriter(ihash, conn);
		return 0;
	}

	long rownum = PQntuples(result);
	if (rownum == 0 )
	{
		PQclear(result);
		sql.str("");
		sql<<"INSERT INTO guid(id, uuid) VALUES(1, "<< ihash<<")";
		DEBUG1(sql.str().c_str());
		PGresult *result = PQexec(conn, sql.str().c_str());
		if(PQresultStatus(result) != PGRES_COMMAND_OK)
		{
			ERRO << PQerrorMessage(conn) << endl;
			PQclear(result);
			pAn->CloseWriter(ihash, conn);
			return 0;
		}
		PQclear(result);
		pAn->CloseWriter(ihash, conn);
		return ihash;
	}
	int i  = atoi(PQgetvalue(result, 0, 0));
	PQclear(result);
	pAn->CloseWriter(ihash, conn);
	return i;
}

bool PostgreStorage :: read_user_value(PGconn *conn, int ihash, const char *uid, const char *dm, const char *key, string &val)
{
	/* 	
		不使用escape因为 domain, key 都已经通过了acl验证，肯定合法
		同时也要求acl必须声明，而非＊匹配 
		*/
	char su[ 2*HERON_KEY_MAXSIZE +1 ];
	int elen =PQescapeString(su, uid, strlen(uid));
	su[elen] = '\0';
   
	ostringstream sql;
	sql <<"SELECT "<< HERON_DB_FIELD_VAL<<" FROM " << HERON_DB_T_HEADER << ihash
            << " WHERE " << HERON_DOMAIN_KEY  <<"='"<< dm <<"' AND "
            << HERON_USRID_KEY <<"='"<< su <<"' AND "<<HERON_DB_FIELD_KEY << "='"<<key<<"'";

DEBUG2("sql", sql.str().c_str());
        PGresult *result = PQexec(conn, sql.str().c_str());
        if(PQresultStatus(result) != PGRES_TUPLES_OK)
        {
		cerr << PQerrorMessage(conn) << endl;
		val = "";
                PQclear(result); 
		return false;
	}

	long rownum = PQntuples(result);
	if (rownum == 0 )
	{
		PQclear(result);
		val = "";
		return false;
	}
	val = PQgetvalue(result, 0, 0);
	PQclear(result);
	return true;
}

int PostgreStorage :: read_user_value(PGconn *conn, int ihash, const char *uid, const char *dm,GList *rkv, GHashTable *kv)
/*
 * rkv和kv 的存在主要是考虑性能，rkv表示要读取的key列表，而kv保存所要读取的数据
 */
{
	/* 	
		不使用escape因为 domain, key 都已经通过了acl验证，肯定合法
		同时也要求acl必须声明，而非＊匹配 
		*/

	char su[ 2*HERON_KEY_MAXSIZE +1 ];
	int elen = PQescapeString(su, uid, strlen(uid));
	su[elen] = '\0';
   
	ostringstream sql, subsql;
	sql <<"SELECT "<< HERON_DB_FIELD_KEY << ", "<< HERON_DB_FIELD_VAL
		<<" FROM " << HERON_DB_T_HEADER << ihash
		<< " WHERE " << HERON_DOMAIN_KEY <<"='"<< dm <<"' AND " << HERON_USRID_KEY 
		<<"='"<< su << "'";

	int j = g_list_length(rkv);
	for(int t = 0; t< j; t++)
	{
		gpointer pd =  g_list_nth_data(rkv, t);
		if(t > 0)
			subsql<<",";           
		subsql << "'"<< (char *)pd <<"'";
	}
	if(j > 0)
		sql << " AND "<< HERON_DB_FIELD_KEY << " IN ("<< subsql.str() << ")";                 
	DEBUG1(sql.str().c_str());
        PGresult *result = PQexec(conn, sql.str().c_str());
        if(PQresultStatus(result) != PGRES_TUPLES_OK)
        {
		cerr << PQerrorMessage(conn) << endl;
                PQclear(result); 
		return -1;
	}

	long rownum = PQntuples(result);
	if (rownum == 0 )
	{
		PQclear(result);
		return 0;
	}
	DEBUG2("result rows= ", rownum);
	for (int i = 0; i < rownum; i++)
	{
		DEBUG3(PQgetvalue(result, i, 0), " = ", PQgetvalue(result, i, 1));
		g_hash_table_insert(kv, g_strdup(PQgetvalue(result, i, 0)), g_strdup( PQgetvalue(result, i, 1)));
	}
	PQclear(result);	
	return rownum;
}


int PostgreStorage :: dump_user_value(PGconn *conn, int ihash, const char *uid, const char *dm, GHashTable *kv)
{
	/* 	
		不使用escape因为 domain, key 都已经通过了acl验证，肯定合法
		同时也要求acl必须声明，而非＊匹配 
		*/

	char su[ 2*HERON_KEY_MAXSIZE +1 ];
	int elen = PQescapeString(su, uid, strlen(uid));
	su[elen] = '\0';
   
	ostringstream sql, subsql;
	sql <<"SELECT "<< HERON_DB_FIELD_KEY << ", "<< HERON_DB_FIELD_VAL
	<<" FROM " << HERON_DB_T_HEADER << ihash
	<< " WHERE " << HERON_DOMAIN_KEY <<"='"<< dm <<"' AND " << HERON_USRID_KEY <<"='"<< su << "'";

	DEBUG1(sql.str().c_str());
        PGresult *result = PQexec(conn, sql.str().c_str());
        if(PQresultStatus(result) != PGRES_TUPLES_OK)
        {
		cerr << PQerrorMessage(conn) << endl;
                PQclear(result); 
		return -1;
	}

	long rownum = PQntuples(result);
	if (rownum == 0 )
	{
		PQclear(result);
		return 0;
	}
	DEBUG2("result rows= ", rownum);
	for (int i = 0; i < rownum; i++)
	{
		DEBUG3(PQgetvalue(result, i, 0)," = ", PQgetvalue(result, i, 1));
		g_hash_table_insert(kv, g_strdup(PQgetvalue(result, i, 0)), g_strdup( PQgetvalue(result, i, 1)));
	}
	PQclear(result);	
	return rownum;
}
